package day0613.managersystem

import akka.actor.{Actor, ActorSystem, Props}
import akka.io.Tcp.Register
import com.typesafe.config.ConfigFactory

import scala.collection.mutable
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global

/**
  * @program: hadoop
  * @description: 主节点
  * @author: Qiang.Ye
  * @date: 2019-06-15 16:42
  **/
class Master extends Actor {

  //保存WorkerID和Worker信息的map
  val idToWorker = new mutable.HashMap[String, WorkerInfo]()

  //保存所有Worker信息的Set
  val workers = new mutable.HashSet[WorkerInfo]()

  //Worker的超时时间
  val WORKER_TIMEOUT = 10 * 1000

  override def preStart(): Unit = {
    //启动定时器
    context.system.scheduler.schedule(5 millis, WORKER_TIMEOUT millis, self, CheckOfTimeOutWorker)
  }


  override def receive: Receive = {

    case RegisterWorker(id, workerHost, memory, cores) => {
      if (!idToWorker.contains(id)) {
        val worker = new WorkerInfo(id, workerHost, memory, cores)
        workers.add(worker)
        idToWorker(id) = worker
        println("new register worker :" + worker)
        sender ! RegisteredWorker(worker.id)
      }
    }
    case SendHeartBeat(workId) => {
      val workerInfo = idToWorker(workId)
      println("get heartbeat message from " + workerInfo)
      workerInfo.lastHeartBeat = System.currentTimeMillis()
    }
    case CheckOfTimeOutWorker => {
      val currentTime = System.currentTimeMillis()
      val toRemove = workers.filter(w => currentTime - w.lastHeartBeat > WORKER_TIMEOUT).toArray

      for (worker <- toRemove) {
        workers -= worker
        idToWorker.remove(worker.id)
      }

      println(" worker size :" + workers.size)


    }

  }
}

object  Master{

  def main(args: Array[String]): Unit = {
    val host = "127.0.0.1"
    val port = 8888
    //创建ActorSystem
    val configStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$host"
         |akka.remote.netty.tcp.port = "$port"
       """.stripMargin
    val config = ConfigFactory.parseString(configStr)
    val actorSystem = ActorSystem.create("MasterActorSystem",config)

    actorSystem.actorOf(Props[Master],"Master")


  }

}
